﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading;
using System.Diagnostics.CodeAnalysis;

namespace Laharsub.Server.Wcf
{
    [SuppressMessage("Microsoft.Design", "CA1001:TypesThatOwnDisposableFieldsShouldBeDisposable",
        Justification = "MemoryStream is disposed in the End method")]
    abstract class PublishMessageAsyncResult : AsyncResult
    {
        static AsyncCallback readCallback = new AsyncCallback(OnReadCompleted);
        byte[] buffer = new byte[256];
        MemoryStream dest = new MemoryStream();
        Stream body;

        protected PubSubMessage Message { get; private set; }

        public PublishMessageAsyncResult(int topicId, string contentType, Stream body, AsyncCallback callback, object state)
            : base(callback, state)
        {
            if (topicId < 1)
            {
                throw new ArgumentOutOfRangeException("topicId", "Topic does not exist.");
            }

            this.body = body;
            this.Message = new PubSubMessage
            {
                ContentType = contentType,
                TopicId = topicId
            };            
        }

        protected void Start()
        {
            this.CopyBufferNext(true);
        }

        public static int End(IAsyncResult result)
        {
            try
            {
                PublishMessageAsyncResult r = AsyncResult.End<PublishMessageAsyncResult>(result);
                return r.Message.MessageId;
            }
            finally
            {
                PublishMessageAsyncResult r1 = result as PublishMessageAsyncResult;
                if (r1 != null && r1.dest != null)
                {
                    r1.dest.Dispose();
                    r1.dest = null;
                }
            }            
        }

        void CopyBufferNext(bool synchronous)
        {
            IAsyncResult result = this.body.BeginRead(this.buffer, 0, buffer.Length, readCallback, this);
            if (result.CompletedSynchronously)
            {
                this.ProcessRead(result, synchronous && result.CompletedSynchronously);
            }
        }

        static void OnReadCompleted(IAsyncResult result)
        {
            if (!result.CompletedSynchronously)
            {
                ((PublishMessageAsyncResult)result.AsyncState).ProcessRead(result, false);
            }
        }

        void ProcessRead(IAsyncResult result, bool synchronous)
        {
            try
            {
                int read = this.body.EndRead(result);
                if (read > 0)
                {
                    this.dest.Write(this.buffer, 0, read);
                    this.CopyBufferNext(synchronous);
                }
                else
                {
                    this.Message.Body = this.dest.ToArray();
                    this.dest.Dispose();
                    this.dest = null;
                    this.buffer = null;
                    this.StartPublish(synchronous);
                }
            }
            catch (Exception e)
            {
                this.Complete(synchronous, e);
            }
        }

        public abstract void StartPublish(bool synchronous);
    }
}
